package cn.itcast.map;

import cn.itcast.bean.SectorBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import cn.itcast.util.DbUtil;
import jdk.nashorn.internal.objects.NativeUint8Array;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.Date;
import java.util.Map;

public class SectorKlineMap extends RichMapFunction<SectorBean, Row> {

    private String kType;
    private String firstTxDateType;

    public SectorKlineMap(String type, String firstTxDateType) {
        this.kType=type;
        this.firstTxDateType=firstTxDateType;
    }

    Long tradeTime=null;
    Long firstTradeTime=null;
    String tradeDate= null;
    Map<String, Map<String, Object>> klineMap=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         * 1.创建构造方法
         *   入参：kType：K线类型
         *         firstTxdate：周期首个交易日
         * 2.获取交易日历表交易日数据
         * 3.获取周期首个交易日和T日
         * 4.获取K线下的汇总表数据（高、低、成交量、金额）
         */

        String sql = "select * from tcc_date where trade_date <= curdate() order by trade_date desc limit 1";
        Map<String, String> dataMap = DbUtil.queryKv(sql);//键为字段名，值为字段对应的时间（yyyy-mm-dd)
        //3.获取周期首个交易日和T日
        tradeDate = dataMap.get("trade_date");
        String firstTradeDate = dataMap.get(firstTxDateType);
        tradeTime = DateUtil.stringToLong(tradeDate, Constant.format_yyyy_mm_dd);
        firstTradeTime = DateUtil.stringToLong(firstTradeDate, Constant.format_yyyy_mm_dd);
        String sqlKline = "select sector_code,\n" +
                "       max(high_price) as high_price,\n" +
                "       min(low_price) as low_price,\n" +
                "       sum(trade_vol) as total_vol,\n" +
                "       sum(trade_amt) as total_amt\n" +
                "from bdp_quot_sector_kline_day\n" +
                "where trade_date between "+firstTradeDate+" and "+tradeDate+" \n" +
                "group by 1;";
        klineMap = DbUtil.query("sector_code", sqlKline);
    }

    @Override
    public Row map(SectorBean sectorBean) throws Exception {
        /**
         * 1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
         * 2.获取T日和周首次交易日时间,转换成long型
         * 3.比较周期首个交易日和当天交易日大小，判断是否是周、月K线
         * 4.获取周/月K数据：成交量、成交额、高、低
         * 5.高、低价格比较
         * 6.计算均价	
         * 7.封装数据Row
         */
        BigDecimal preClosePrice = sectorBean.getPreClosePrice();
        BigDecimal closePrice = sectorBean.getClosePrice();
        BigDecimal openPrice = sectorBean.getOpenPrice();
        BigDecimal highPrice = sectorBean.getHighPrice();
        BigDecimal lowPrice = sectorBean.getLowPrice();
        Long tradeVolDay = sectorBean.getTradeVolDay();
        Long tradeAmtDay = sectorBean.getTradeAmtDay();
        if(firstTradeTime<tradeTime && (kType.equals("2") || kType.equals("3"))){
            //成交量、成交额、高、低
            Map<String, Object> map = klineMap.get(sectorBean.getSectorCode());
            if(map!=null && map.size()>0){
                Long totalVol = Long.valueOf(map.get("total_vol").toString());
                Long totalAmt = Long.valueOf(map.get("total_amt").toString());
                tradeVolDay += totalVol;
                tradeAmtDay += totalAmt;

                BigDecimal high_price = new BigDecimal(map.get("high_price").toString());
                BigDecimal low_price = new BigDecimal(map.get("low_price").toString());

                if(high_price.compareTo(highPrice)==1){ //>
                    highPrice = high_price;
                }
                if(low_price.compareTo(lowPrice)==-1){
                    lowPrice = low_price;
                }
            }
        }
        BigDecimal avgPrice = new BigDecimal(0);
        if(tradeVolDay!=0){
            avgPrice = new BigDecimal(tradeAmtDay).divide(new BigDecimal(tradeVolDay),2, RoundingMode.HALF_UP);
        }
        Row row = new Row(13);
        row.setField(0,new Timestamp(new Date().getTime())); //update_time
        row.setField(1,tradeDate); //trade_date
        row.setField(2,sectorBean.getSectorCode()); //sector_code
        row.setField(3,sectorBean.getSectorName()); //sector_name
        row.setField(4,kType); //k_line_type
        row.setField(5,preClosePrice); //pre_close_price
        row.setField(6,openPrice); //open_price
        row.setField(7,highPrice); //high_price
        row.setField(8,lowPrice); //low_price
        row.setField(9,closePrice); //close_price
        row.setField(10,avgPrice); //avgprice
        row.setField(11,tradeVolDay); //trade_vol
        row.setField(12,tradeAmtDay); //trade_amt

        return row;
    }
}
